#include "SocketTcpClientThreads.hpp"
#include "cybertron/core/Log.hpp"
#include "cybertron/core/UtilCRT.hpp"
#include "cybertron/network/SocketBase.hpp"

CYBERTRON_BEGIN

SocketTcpClientThreads::SocketTcpClientThreads(const std::string& name)
{
	mName = name;
	mThreadRunning = false;
	mHeartbeatSeconds = 5;
}

SocketTcpClientThreads::~SocketTcpClientThreads()
{
	stop();
}

void SocketTcpClientThreads::setupContext()
{
	mpIoContext = std::make_shared<asio::io_context>();
}

void SocketTcpClientThreads::start(int numThreads)
{
	if (mThreads.size() > 0)
	{
		throw std::runtime_error("Stop context thread first!");
	}

	auto pContext = mpIoContext;
	unsigned int numCores = std::thread::hardware_concurrency();
	if (numThreads < 0)
	{
		numThreads = (int)(numCores);
	}

	mThreadRunning = true;

	for (int i = 0; i < numThreads; ++i)
	{
		std::string name = mName + " IO";

		auto pThread = UtilCRT::startThread(
			name.c_str(), 
			[this, pContext]()
		{
			while (mThreadRunning)
			{
				if (pContext != nullptr)
				{
					pContext->run();
				}
			}
		});

		mThreads.push_back(pThread);
	}

	std::string name = mName + " Heartbeat";
	mHeartbeatThread = UtilCRT::startThread(
		name.c_str(),
		[this]()
	{
		onHeartbeat();
	});
}

void SocketTcpClientThreads::addSocket(SocketBasePtr pSocket)
{
	if (pSocket == nullptr)
	{
		return;
	}

	std::unique_lock<std::recursive_mutex> lock(mHeartbeatMutex);
	for (size_t i = 0; i < mClientSockets.size(); ++i)
	{
		auto pExisting = mClientSockets[i].lock();
		if (pExisting == pSocket)
		{
			return;
		}
	}

	mClientSockets.push_back(pSocket);
}

void SocketTcpClientThreads::removeSocket(SocketBase* pSocket)
{
	if (pSocket == nullptr)
	{
		return;
	}

	std::unique_lock<std::recursive_mutex> lock(mHeartbeatMutex);
	for (size_t i = 0; i < mClientSockets.size(); ++i)
	{
		auto pExisting = mClientSockets[i].lock();
		if (pExisting.get() == pSocket)
		{
			mClientSockets.erase(mClientSockets.begin() + i);
			return;
		}
	}
}

void SocketTcpClientThreads::onHeartbeat()
{
	unsigned int fps = 30;
	unsigned int sleep_milliseconds = 0;

	while (mThreadRunning)
	{
		std::chrono::milliseconds ms = std::chrono::milliseconds(1000 / 30);
		std::this_thread::sleep_for(ms);

		sleep_milliseconds += (unsigned int)(ms.count());

		if (sleep_milliseconds < mHeartbeatSeconds * 1000)
		{
			continue;
		}

		sleep_milliseconds = 0;

		std::vector<SocketBasePtr> sockets;
		{
			std::unique_lock<std::recursive_mutex> lock(mHeartbeatMutex);
			for (size_t i = 0; i < mClientSockets.size(); ++i)
			{
				auto pSocket = mClientSockets[i].lock();
				if (pSocket != nullptr)
				{
					sockets.push_back(pSocket);
				}
				else
				{
					mClientSockets.erase(mClientSockets.begin() + i);
					--i;
				}
			}
		}

		for (size_t i = 0; i < sockets.size(); ++i)
		{
			sockets[i]->sendHeartbeat();
		}
	}
}

void SocketTcpClientThreads::stop()
{
	mThreadRunning = false;

	if (mpIoContext != nullptr)
	{
		mpIoContext->stop();
	}

	for (size_t i = 0; i < mThreads.size(); ++i)
	{
		mThreads[i]->join();
	}

	mThreads.clear();

	if (mHeartbeatThread != nullptr)
	{
		mHeartbeatThread->join();
		mHeartbeatThread = nullptr;
	}

	{
		std::unique_lock<std::recursive_mutex> lock(mHeartbeatMutex);
		mClientSockets.clear();
	}
}

CYBERTRON_END
